/*
Copyright (c) 2025, 北京智慧物联科技有限公司.

简体中文
保留所有权利。依据GPL 2.0（以下简称"本许可"）进行许可。除非符合（遵守）本许可及以下条款，否则您不得使用此文件：

禁止任何未经授权的个人或组织以任何形式将本程序及其附属资料、创新、创意、架构设计、算法、衍生作品等用于任何商业目的、盈利活动、各类竞赛(比赛)、直播教学、录播教学、线下课程、书籍编写、教材编写、会议、培训、公益活动、项目课题、毕业设计、毕业论文、学术论文等。

本程序由版权持有人和贡献人“按原样”提供，不提供任何明示或暗示担保，包括但不局限于对适销性和特定用途适合性的暗示担保。在任何情况下，版权持有人或贡献人对因使用本程序而导致的任何直接、间接、附带、特殊、典型或后果性损害（包括但不限于购买替代商品或服务；使用损失、数据丢失或利润损失；业务中断）不承担任何责任，无论其如何产生，也不论其责任理论为何，无论是合同、严格责任还是侵权（包括疏忽或其他），即使已告知此类损害的可能性。

English
All rights reserved. Licensed under the GPL 2.0 (the "License"). You may NOT use this file except in compliance with the License and the following terms:

IT IS PROHIBITED FOR ANY UNAUTHORIZED PERSON OR ENTITY TO USE THIS PROGRAM AND ITS ANCILLARY DATA, INNOVATIONS, CONCEPTS, ARCHITECTURE DESIGN, ALGORITHMS, DERIVED WORKS, ETC., IN ANY FORM FOR COMMERCIAL PURPOSES, PROFIT-MAKING ACTIVITIES, VARIOUS COMPETITIONS (CONTESTS), LIVE TEACHING, PRE-RECORDED LESSONS, OFFLINE COURSES, BOOK WRITING, CONFERENCES, TRAININGS, PUBLIC WELFARE ACTIVITIES, PROJECTS, TOPICS, GRADUATION DESIGNS, DISSERTATIONS, THESES, AND OTHER APPLICATIONS.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <time.h>

#include "ohos_init.h"
#include "cmsis_os2.h"

#include "lwip/inet.h"
#include "lwip/sockets.h" 
#include <netdb.h>      // 替代 wm_ntp.h?

#include <cJSON.h> 

#include "iot_gpio.h"   // OpenHarmony HAL API：IoT硬件设备操作接口-GPIO
#include "iot_gpio_ex.h"
// include
#include "iot_sha256.h"
#include "iot_string.h"  
#include "iot_time.h"            
#include "libemqtt.h" 

#include "uart_task.h"          // radar_start_task 
#include "station_demo.h"

#include "mqtt_main.h"          // mqtt 相关参数
#include "mqtt_down.h"           // parseReceivedJson,  gpio_init, led_on

#include "soc_osal.h"
#include "errcode.h"  

// 变量
static uint8_t packet_buffer[MQTT_DEMO_RECV_BUF_LEN_MAX];
static int socket_id;
static int mqtt_keepalive = 20; //60;
static mqtt_broker_handle_t mqtt_broker;  // mqtt broker   

/***  iot 参数  ***/  
osThreadId_t g_mqtt_id, g_ping_id;   // 线程ID 

char tm_h[10];       // yyyymmddhh
char mac[13] = {0};  // device id  
char client_id[64];  // dev_mac_tm: dev_14cf92c2aa32_2021082220, dev_ac923222b24b_2021082220
char password[64];   // sha256(tm + key) =>
char server_host[64] = {0} ;   //  
char server_port[64] = {0} ;   //  
char *topic_pub;
char *topic_sub;
char *topic_ack;
char *topic_will;
  
static int g_mqtt_ok = 0;       // mqtt连接标志，1完成, -1未完成；
int g_init_ok = 0;              // parameter标志，1正常，其他值不正常
int g_publish_fail_count = 0;   // 发布失败次数
time_t g_publish_fail_time;

extern int g_ConnectSuccess;    // wifi连接标志，1成功，0未连接；引用了 wifi_connect.c 中的同名变量

// neptune: 5U=10ms,500U=1s      
//  hi3861: 1U=10ms,100U=1s
 
/***  初始化函数  ***/  
 
// 设备初始化，在网络连接之后，读取 mac address, tm, 并加密形成 password
// https://www.cnblogs.com/dpf-learn/p/7580568.html
static int Parameter_Init(void) {
    osal_printk("\r\n[Init 2] Parameter Init ... \r\n");  

    getNtpTime();                       // 获取网络授时
	osDelay(5*100);                     // 3*1000，大概3秒返回时间值

    currTimeHour(tm_h);                 // 时间戳，精确到小时 

    GetDeviceMac(mac);                  // 读取mac地址
    if(mac[0]==0) {
        osal_printk("=!= Can not get mac address, stop here. \r\n"); 
        return 0;
    } 
    osal_printk("=^= mac: %s\r\n", mac); 

    // dev_ac923222b24b_YYYYMMDDHH
    sprintf(client_id, "dev_%s_%s", mac, tm_h);
    osal_printk("=^= client_id: %s\r\n", client_id);

    unsigned char *result; 
    result = HMAC_SHA256(tm_h, MQTT_DEMO_HMAC_KEY);  // 32位，UC

    unCharToStr(password, result, 32);               // 转换成 64位字符
    osal_printk("=^=  password: %s\r\n", password);  
 
    // 属性上报主题，使用时间 device_id 替换："$dev/{device_id}/rpt/up" 
    topic_pub = str_replace( MQTT_DEMO_PUB_TOPIC, "{device_id}", mac);
    osal_printk("=^= topic_pub: %s\r\n", topic_pub); 
 
    // "$dev/{device_id}/cmd/down/#"
    topic_sub = str_replace( MQTT_DEMO_SUB_TOPIC, "{device_id}", mac);
    osal_printk("=^= topic_sub: %s\r\n", topic_sub); 

    // $dev/{device_id}/cmd/up/{request_id}
    topic_ack = str_replace( MQTT_DEMO_ACK_TOPIC, "{device_id}", mac);
    osal_printk("=^= topic_ack: %s\r\n", topic_ack); 

    // $dev/{device_id}/will 
    topic_will = str_replace( MQTT_DEMO_WILL_TOPIC, "{device_id}", mac);
    osal_printk("=^=topic_will: %s\r\n", topic_will);

    // 从nv中读取 mqtt 参数
    //Read_Mqtt(server_host, server_port); 
     
    if(server_host[0]==0 || server_port[0]==0){     
        printf("=!= flash's host is null. \r\n");
        sprintf(server_host, "%s", MQTT_DEMO_SERVER_ADDR);           // 使用默认的mqtt参数
        sprintf(server_port, "%s", MQTT_DEMO_SERVER_PORT); 
    }
    osal_printk("==< host: %s, port: %s.\r\n", server_host, server_port); 

    osDelay(10*100);  // 延迟连接时间，必须大于 will 时间20秒，
     
    // LED灯闪烁5次
    LED_Spark(5, 1);

    return 1;
}
  
/***  心跳机制  ***/ 

int pingCnt = 0; 

// 不能使用定时期，因为在timer中不能有阻塞任务mqtt_ping
void Ping_Run(void) {     
    while(1) {
        pingCnt++;
        //printf("=>> ping %d ...\r\n", pingCnt); 
        mqtt_ping(&mqtt_broker);    // 阻塞任务，在timer中不能使用
        osDelay(5 * 100U);          // 5 * 1000ms 
    }
}
 
// 启动定时器任务线程 
//static 
void ping_start_task(void){
    osThreadAttr_t attr; 
    attr.attr_bits = 0U;
    attr.cb_mem = NULL;
    attr.cb_size = 0U;
    attr.stack_mem = NULL;
    attr.stack_size = 4096;
    attr.priority = osPriorityNormal;    
	attr.name = "Ping_Run"; 
    g_ping_id = osThreadNew((osThreadFunc_t)Ping_Run, NULL, &attr);
    if (g_ping_id== NULL) {
        printf("=!= ping_start_task() ... falied to create Ping_Run.\n");
    }
}

// 终止进程，停止检测
//static 
void ping_stop_task(void) {   
    if(g_ping_id) {
	    osStatus_t status = osThreadTerminate(g_ping_id);
        printf("### ping_stop_task() ... status: %d.\r\n", status);
    }
}

/***  下发命令函数  ***/ 
// 移至 mqtt_cmd.c 中

/***  mqtt 基础函数  ***/ 

static int close_socket(mqtt_broker_handle_t *broker)
{
    int fd = broker->socketid;
    return closesocket(fd);
}

static int send_packet(int socket_info, const void *buf, unsigned int count)
{
    int fd = socket_info;
    return send(fd, buf, count, 0);
}


// struct timeval *timeout是select的超时时间，这个参数至关重要，它可以使select处于三种状态:
// 第一，若将NULL以形参传入，即不传入时间结构，就是将select置于阻塞状态，一定等到监视文件描述符集合中某个文件描述符发生变化为止；
// 第二，若将时间值设为0秒0毫秒，就变成一个纯粹的非阻塞函数，不管文件描述符是否有变化，都立刻返回继续执行，文件无变化返回0，有变化返回一个正值；
// 第三，timeout的值大于0，这就是等待的超时时间，即select在timeout时间内阻塞，超时时间之内有事件到来就返回了，否则在超时后不管怎样一定返回，返回值同上述。
// https://blog.csdn.net/zhaodeming000/article/details/98618097

static int read_packet(int sec, int us)
{ 
/*  by:liuyp@20241222, 在oh4下无法返回有效值，注释掉。
    int ret = 0;
    if ((sec >= 0) || (us >= 0))
    {
        fd_set readfds;
        struct timeval tmv;

        // 初始化文件描述符集
        FD_ZERO (&readfds);
        FD_SET (socket_id, &readfds);

        // 初始化超时数据结构
        tmv.tv_sec = sec;
        tmv.tv_usec = us;

        // 如果超时，1如果输入可用，则选择返回0，如果错误，则为1
        ret = select(socket_id + 1, &readfds, NULL, NULL, &tmv);
        if(ret < 0)
            return -2;
        else if(ret == 0)
            return MQTT_DEMO_READ_TIMEOUT;

    }
*/
    
    int total_bytes = 0, bytes_rcvd, packet_lens;
    memset(packet_buffer, 0, sizeof(packet_buffer));

    // 真正的接受数据地方
    if((bytes_rcvd = recv(socket_id, (packet_buffer + total_bytes), MQTT_DEMO_RECV_BUF_LEN_MAX, 0)) <= 0)
    {
        // wifi 断联后收数为 -1
        //printf("### recv error: len = %d,  socket = %d. \r\n", bytes_rcvd, socket_id);
        return -1;
    }

    total_bytes += bytes_rcvd; // 保持总字节的计数
    if (total_bytes < 2)
        return -1; 

    uint16_t rem_len = mqtt_parse_rem_len(packet_buffer);
    uint8_t rem_len_bytes = mqtt_num_rem_len_bytes(packet_buffer);

    packet_lens = rem_len + rem_len_bytes + 1;

    while(total_bytes < packet_lens) // 读数据包
    {
        if((bytes_rcvd = recv(socket_id, (packet_buffer + total_bytes), MQTT_DEMO_RECV_BUF_LEN_MAX, 0)) <= 0)
            return -1;
        total_bytes += bytes_rcvd; // 保持总字节的计数
    }

    return packet_lens;
}

static int init_socket(mqtt_broker_handle_t *broker, const char *hostname, short port)
{  
    // 创建套接字
    if((socket_id = socket(AF_INET, SOCK_STREAM, 0)) < 0)  {
        printf("=^= init_socket: create socket failed\r\n"  );
        return -1;
    } 

    // 初始化预连接的服务端地址
    struct sockaddr_in socket_address;
    memset(&socket_address, 0, sizeof(struct sockaddr_in));

    socket_address.sin_family = AF_INET;
    socket_address.sin_port = htons(port); 
    socket_address.sin_addr.s_addr = inet_addr(hostname);

    // 连接套接字
    int ret = connect(socket_id, (struct sockaddr *)&socket_address, sizeof(socket_address));
    if (ret < 0){
        printf("=!= connect failed. ret = %d \r\n", ret);  
        close_socket(&mqtt_broker);
        return -1;
    }
    osDelay(10);

    // MQTT stuffs 
    //broker->alive = mqtt_keepalive;
    broker->socketid = socket_id;
    broker->mqttsend = send_packet; 
    //printf("=^= init_socket: end. \r\n"); 

    return 0;
}
 
/***  mqtt 相关函数  ***/ 

// 订阅主题
static int subscribe_topic(char *topic)    {
    unsigned short msg_id = 0, msg_id_rcv = 0;
    int packet_lengthgth = 0;
	int ret = -1;
 
    if(topic == NULL) {
        return -1;
    }
    
    ret = mqtt_subscribe(&mqtt_broker, topic, &msg_id);
	if( ret == -1 ) {
		close_socket(&mqtt_broker);
		return -1;
	}
    
    packet_lengthgth = read_packet(MQTT_DEMO_READ_TIME_SEC, MQTT_DEMO_READ_TIME_US);
    if(packet_lengthgth < 0)
    {
        printf("Error(%d) on read packet!\n", packet_lengthgth);
        close_socket(&mqtt_broker);
        return -1;
    }
 
    if(MQTTParseMessageType(packet_buffer) != MQTT_MSG_SUBACK)
    {
        printf("SUBACK expected!\n");
        close_socket(&mqtt_broker);
        return -2;
    }
 
    msg_id_rcv = mqtt_parse_msg_id(packet_buffer);
	if(msg_id != msg_id_rcv)
	{
		printf("%d message id was expected, but %d message id was found!\n", msg_id, msg_id_rcv);
        close_socket(&mqtt_broker);
        return -3;
	}
 
    return 0;
}

// 解析接收到的数据 
// 移至 mqtt_cmd.c

// 响应指令组包
static int packPublishAck(char *jsonBuffer)
{
    cJSON *jsRet = NULL;
    int ackLen = 0;
    jsRet=cJSON_CreateObject(); 

    if(jsRet) {
        cJSON_AddStringToObject(jsRet,"result_code","0");
        //cJSON_AddStringToObject(jsRet, "response_name", "led_control");
        //cJSON *jsArray=cJSON_CreateObject();
        //cJSON_AddItemToObject(jsRet, "paras", jsArray);
        //cJSON_AddStringToObject(jsArray, "result", "success");
        
        char *databuf = cJSON_PrintUnformatted(jsRet);
		if(databuf) {
			//printf("json: %s\r\n", databuf);
			if( jsonBuffer ) {
				ackLen = strlen(databuf);
				memcpy(jsonBuffer, databuf, ackLen);
			}
            tls_mem_free(databuf);
        }
        cJSON_Delete(jsRet); 
    }
    return ackLen;
}


// 上线信息组包
static int packOnline(char *jsonBuffer)
{ 
    int ackLen = 0;
    cJSON *jsRet = NULL;
    jsRet = cJSON_CreateObject();

    if(jsRet) {
        cJSON_AddStringToObject(jsRet, "state", "online"); 

        char *databuf = cJSON_PrintUnformatted(jsRet);
		if(databuf) {
			//printf("json: %s\r\n", databuf);
			if( jsonBuffer ) {
				ackLen = strlen(databuf);
				memcpy(jsonBuffer, databuf, ackLen);
			}
            tls_mem_free(databuf);
        }
        cJSON_Delete(jsRet); 
    }
    return ackLen;
}

// 发送上线信息
static int publishOnline(void){ 
    int32_t ackLen = 0;
	char *ackBuffer = NULL; 
    ackBuffer = tls_mem_alloc(MQTT_DEMO_RECV_BUF_LEN_MAX);

    if( ackBuffer )  {
        memset(ackBuffer, 0, MQTT_DEMO_RECV_BUF_LEN_MAX);
        ackLen = packOnline(ackBuffer);      // 组包
        
        // 一个Topic只能有1条Retained消息，发布新的Retained消息将覆盖老的Retained消息 
        // 即覆盖之前的 will (state: offline)                                  
        mqtt_publish(&mqtt_broker, topic_will, ackBuffer, ackLen, 1);       // 发送 will 主题，retain保留位
        //osal_printk("==> publishOnline return: %d \r\n", ret);

        tls_mem_free(ackBuffer);
        return 1;
    }
    else {
        return -1;
    }
}
  
// *** 发布雷达报文 ***

int publish_sle_data(const uint8_t *str)
{
    // publish_radar_data(cmd, value, result, action, timestamp);  // 发布数据 
    // 8003_01_1_moving parameter: 1
    char *token;
    char *delim ="_";
    char *sarr[4];
    token = strtok(str, delim);
    int i=0;
    while(token!=NULL){
        sarr[i++] = token;
        token = strtok(NULL, delim);
    }

    char *cmd = sarr[0];
    char *value = sarr[1];
    char *result = sarr[2];
    char *action = sarr[3];
    char *timestamp = NULL;

    int ret =  publish_radar_data(cmd, value, result, action, timestamp);  // 发布数据  
    osal_printk("=*= publish_sle_data: %s \r\n", str);
    return ret;
}

// 组成报文包，并发布
int publish_radar_data(const char *cmd, const char *value, 
    const char *result, const char *action, char *timestamp)
{
    /*
    if(g_ConnectSuccess==0){
        printf("=!= Wifi disconnect, can't publish. \r\n");
        return 0;           //未联网，则停止后续
    }*/

    // 如果设备未联网，则不发布
    if(g_mqtt_ok != 1){
        osal_printk("=!= mqtt server disconnect, can't publish. \r\n");
        return 0;           //未联网，则停止后续
    }

    // by:liuyp@20241221
    //currTimeStamp(timestamp);   // 当前时间 
    timestamp = "20250105T141516Z";

    int ackLen = 0;

	cJSON *jsRet = NULL;
	cJSON *jsArray = NULL;
    cJSON *arrayObj_1 = NULL;
    cJSON *arrayObj_2 = NULL;
    
    jsRet = cJSON_CreateObject();
	if(jsRet) {   
		jsArray = cJSON_CreateArray();
        cJSON_AddItemToObject(jsRet, "services", jsArray); 
        { 
            arrayObj_1 = cJSON_CreateObject();
            cJSON_AddStringToObject(arrayObj_1, "service_id", MQTT_SERVICE_ID);  // HumanHealth 
            
            arrayObj_2 = cJSON_CreateObject();
            {
                cJSON_AddStringToObject(arrayObj_2, "cmd", cmd);
                cJSON_AddStringToObject(arrayObj_2, "value", value);
                cJSON_AddStringToObject(arrayObj_2, "result", result);
                cJSON_AddStringToObject(arrayObj_2, "action", action);  
            } 
            cJSON_AddItemToObject(arrayObj_1, "properties", arrayObj_2); 
            cJSON_AddStringToObject(arrayObj_1,"event_time", timestamp);   

            cJSON_AddItemToArray(jsArray, arrayObj_1);  // 加到 services 所在的 jsArray中 
        }

        char *databuf = cJSON_PrintUnformatted(jsRet);
        //{"services":[{"service_id":"HumanRadar","properties":
        //    {"cmd":"8102","value":"49","result":"73","action":"heart: 73 bpm"},
        //    "event_time":"20220414T165708Z"}]}

		if(databuf) { 
			ackLen = strlen(databuf); 
			
            // 发布
            int ret = mqtt_publish(&mqtt_broker, (const char *)topic_pub, databuf, ackLen, 0);
		
            // 释放内存
            tls_mem_free(databuf); 
            cJSON_Delete(arrayObj_2);
            cJSON_Delete(arrayObj_1);
            cJSON_Delete(jsArray);
            cJSON_Delete(jsRet);

            if( ret!=1 ){
                osal_printk("!!! error: publish failed... \r\n");         

                // 连续3次发送不成功，说明mqtt服务出问题了，需要重新连接
                /*
                g_publish_fail_count ++ ;
                g_publish_fail_time = time(NULL);  // 单位为s

                if(g_publish_fail_count>3){

                    time_t currTime = time(NULL);
                    double df = difftime(currTime , g_publish_fail_time);   // 单位 s
                    (void)df;
                    //if(df <=5)        // 如果 5 秒内
                    {   
                        g_publish_fail_count = 0;

                        //ping_stop_task();               // 关闭ping定时器，需要写入nv日志
                        //close_socket(&mqtt_broker);     // 关闭mqtt连接

                        // Mqtt_Start();                   // 重新启动 mqtt 服务，不允许再次重连，因为可能wifi断了。
                    }
                }
                */
            }
        } 
        
        cJSON_Delete(arrayObj_2);
        cJSON_Delete(arrayObj_1);
        cJSON_Delete(jsArray);
        cJSON_Delete(jsRet);
    }
    return ackLen;
}

/***  mqtt 初始化函数  ***/

int mqttCnt = 0;  // mqtt 连接次数，3次(每次10秒)之后改用默认的 mqtt参数连接(248)

// 1成功，其他值失败
//static  
int Mqtt_Start(void)
{
    osDelay(10);
    int packet_lens,  ret = 0; 
 
    osal_printk("\r\n[Init 3] Start MQTT ... \r\n");  
    
    osal_printk("step1: init mqtt lib... \r\n");                //步骤1:MQTT初始化 
    mqtt_init(&mqtt_broker, client_id);

    osal_printk("step2: establishing tcp connection... \r\n");  //步骤2:建立TCP连接   
    mqtt_init_auth(&mqtt_broker, (char*)client_id, (char*)password); 
    
    //ret = init_socket(&mqtt_broker, MQTT_DEMO_SERVER_ADDR, MQTT_DEMO_SERVER_PORT);   // 使用从nv中读取的参数
    ret = init_socket(&mqtt_broker, server_host, atoi(server_port) ); 
    if(ret != 0) { 
        // 如果mqtt server未启动，则错误出现在这里
        osal_printk("=!= init_socket failed. ret = %d \r\n", ret); 
        mqttCnt++;
        return -1;
    }

    // set will topic:
    mqtt_set_alive(&mqtt_broker, mqtt_keepalive);

    // will_flag =1
    mqtt_set_will(&mqtt_broker, 1, topic_will, MQTT_DEMO_WILL_MSG);   

    osal_printk("step3: establishing mqtt connection... \r\n");     //步骤3:建立MQTT连接 
    ret = mqtt_connect(&mqtt_broker); 
    if(ret) {
        osal_printk("=!= mqtt_connect fail. ret=%d\r\n", ret);
        return -2;
    } 

    //packet_lens = read_packet(MQTT_DEMO_READ_TIME_SEC, MQTT_DEMO_READ_TIME_US);
    packet_lens = read_packet(0, 0);
    if(packet_lens < 0) {
        osal_printk("=!= Error(%d) on read packet!\n", packet_lens);
        close_socket(&mqtt_broker);
        return -3;
    }
    //printf("0x%02x\n",packet_buffer[3]);
    if(MQTTParseMessageType(packet_buffer) != MQTT_MSG_CONNACK || packet_buffer[3] != 0x00) {
        osal_printk("=!= CONNACK expected or failed! %d\n", MQTTParseMessageType(packet_buffer));
        close_socket(&mqtt_broker);
        return -4;
    }


    osal_printk("step4: subscribe mqtt... %s \r\n", topic_sub);                    //步骤4：订阅MQTT 
    if(subscribe_topic(topic_sub) != 0) {
        osal_printk("=!= subscribe topic failed. \r\n");
        return -5;
    }

    // 自定义的其他动作：
    
    // 4.1 online
    osDelay(10);
    //printf("^=^ MQTT Connect & subscribe topic success. \r\n");    
    publishOnline();  // 此处发送 online也未必 不会再收到自己的 offline 的will  

    osal_printk("step5: start the ping timer... \r\n");               //步骤5：启动心跳进程
    ping_start_task(); 

    mqttCnt = 0;    // 成功了则归0
    return 1;       // 1表示成功
}

// 接收来自服务端的数据
//static 
int Mqtt_Receive(void) {
    int packet_lens = 0; //, ret = 0;
    packet_lens = read_packet(0, 1);   // 数据已经保存在 packet

    if(packet_lens == MQTT_DEMO_READ_TIMEOUT) {
        return MQTT_DEMO_READ_TIMEOUT;      // 超时
    }
    else if(packet_lens > 0) {
        if (MQTTParseMessageType(packet_buffer) == MQTT_MSG_PUBLISH) {
            //printf("<<= Mqtt_Receive. \r\n");

            // 1. 解包 topic
            int len, len2;
            char topic[64 + 16] = {0}, request_id[48] = {0}, ACK_TOPIC[128] = {0};   

            len = mqtt_parse_pub_topic(packet_buffer, (uint8_t*)topic);                   
			topic[len] = '\0';

            osal_printk("<<= topic_sub: %s. \r\n", topic); 
            len2 = strlen(topic_sub)-1;                 // 去掉 ...down/# 的 # 字符
            strncpy(request_id, topic+len2, len-len2);  // 取出 down/123 的 123
            //printf("= request_id: %s. \r\n ", request_id);
  
            //printf("= topic_ack: %s%s. \r\n", topic_ack, request_id);
            sprintf(ACK_TOPIC, "%s%s", topic_ack, request_id);  // 使用转换后的 topic_ack / request_id

            // 2. 立即回复响应报文
            int ackLen = 0;
			char *ackBuffer = NULL;
            ackBuffer = tls_mem_alloc(MQTT_DEMO_RECV_BUF_LEN_MAX);
            if( ackBuffer )  {
                memset(ackBuffer, 0, MQTT_DEMO_RECV_BUF_LEN_MAX);
				ackLen = packPublishAck(ackBuffer);                             // 组包
				mqtt_publish(&mqtt_broker, ACK_TOPIC, ackBuffer, ackLen, 0);    // 发送
				tls_mem_free(ackBuffer);   

                //printf("= mqtt_publish:   \r\n" );
            }
            else {
                tls_mem_free(ackBuffer);   
                osal_printk("<<= packPublishAck:  ackBuffer is null \r\n" );
                return -1;
            } 

            // 3. 解析收到的数据
            uint8_t *msg;
            msg = malloc(MQTT_DEMO_RECV_BUF_LEN_MAX);
            memset(msg, 0, MQTT_DEMO_RECV_BUF_LEN_MAX);
            int len9 = mqtt_parse_publish_msg(packet_buffer, &msg); // 解包 msg

            // 定义在 mqtt_cmd.h 中：如果 len9 = -1
            //printf("<<= before parseReceivedJson. \r\n");
            len9 = parseReceivedJson(msg);   // (void)len9;
            if(len9==-1){
                free(msg);  
                return -3;      
            }
            free(msg);   
        }
        else if(MQTTParseMessageType(packet_buffer) == MQTT_MSG_PINGRESP) {
            //printf("<<= recv pong. \r\n");
        }
        else {
            printf("<<= unknow mqtt msg type: %d \r\n", MQTTParseMessageType(packet_buffer));// packet_buffer[0]);
        }
        return 1;  
    }
    else if(packet_lens == -1)  {
    	//printf("packet_lens:%d\n", packet_lens);
        printf("=!= mqtt error (len = %d), stop mqtt! \r\n", packet_lens); 

        ping_stop_task();               // 关闭ping定时器，需要写入nv日志
        close_socket(&mqtt_broker);     // 关闭mqtt连接 
    
        return -2;
    }   
    return 1;
}

// mqtt 主程序
void Mqtt_Main(void) {  
    // wifi connect 成功后进入本方法

    // init some parameters, iot 参数和签名, topics
    g_init_ok = Parameter_Init();
    if( g_init_ok == 0 ) {
        return;   // mac is null
    }

start_m:     
    // mqtt start
    g_mqtt_ok = Mqtt_Start();
    osDelay(10);        

    if( g_mqtt_ok == 1 )  { 
        g_publish_fail_count = 0;  

        // 1. app update
        read_hi3861_version();            // = Firmware_Version, 如果设备版本比服务器低，则后台下发 Firmware_Ota 指令
        osDelay(10);                      // 100ms，    
        
        // 2. chip update
        Radar_Version();                  // = Radar_Version，如果设备版本比服务器低，则后台下发 Radar_Ota 指令    
        osDelay(10);                      // 100ms，

        // 死循环接收 mqtt server 端的指令
        while (1){ 
            Mqtt_Receive();         // 死循环接收mqtt后台指令，包括ping回应数据、后台指令 
            osDelay(10);            // 10U = 100ms = 0.1s
        } 
    } 
    else if( g_mqtt_ok == -1){      // 假如nv中存储的mqtt参数连接不成功，则使用默认的参数，即连接到248服务器
        if( mqttCnt >=3 ){
            // 连接不成功后，使用默认的 mqtt 参数 
            sprintf(server_host, "%s", MQTT_DEMO_SERVER_ADDR);           
            sprintf(server_port, "%s", MQTT_DEMO_SERVER_PORT);   
            printf("==< host: %s, port: %s.\r\n", server_host, server_port);   
             
            goto stop_m;                     // 重新启动 mqtt 服务
        }
        else { 
            goto stop_m;                     // 重新启动 mqtt 服务
        }
    }

stop_m:
    // server 端响应不及时易造成 重复连接，加大延迟时长为10秒。
    osDelay(10 * 100U);             // 延迟3秒 -> 改成10秒，2023-06-12
    close_socket(&mqtt_broker);     // 关闭mqtt连接
    goto start_m;                   // 重新启动 mqtt 服务 
    
}


/***  程序入口  ***/ 

// app entry  
void mqtt_start_task(void)
{  
    osDelay(100);
    osThreadAttr_t attr; 
    attr.attr_bits = 0U;
    attr.cb_mem = NULL;
    attr.cb_size = 0U;
    attr.stack_mem = NULL;
    attr.stack_size = 4096;
    attr.priority = osPriorityNormal; 
    attr.name = "Mqtt_Main";
    g_mqtt_id = osThreadNew((osThreadFunc_t)Mqtt_Main, NULL, &attr);
    if (g_mqtt_id == NULL)  {
        osal_printk("=!= mqtt_start_task() ... falied to create Mqtt_Main!\n");
    }
}

// 终止 mqtt 线程 
void mqtt_stop_task(void) {   
    if(g_mqtt_id) {
        //!!! mqtt无权停止雷达报数！radar_stop_task();              // 雷达停止报数
        ping_stop_task();               // 关闭ping定时器
        close_socket(&mqtt_broker);     // 关闭mqtt连接 

	    osStatus_t status = osThreadTerminate(g_mqtt_id);
        osal_printk("### mqtt_stop_task() ... status: %d.\r\n", status);
    }
}
 
// 程序入口 
//APP_FEATURE_INIT(mqtt_start_task); 
